package com.hujiang.basic.search.handler.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.hujiang.basic.framework.core.util.IpUtil;
import com.hujiang.basic.framework.plugin.mq.delivery.MessageType;
import com.hujiang.basic.framework.rest.common.HttpClientUtil;
import com.hujiang.basic.search.common.exception.ConsumeMessageException;
import com.hujiang.basic.search.common.service.NotifyService;
import com.hujiang.basic.search.fullindex.TaskDao;
import com.hujiang.basic.search.model.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.MDC;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class ForwardingHandler implements MessageListener {

    @Autowired
    private IndexHandler indexer;

    @Autowired
    private DeleteHandler deleter;

    @Autowired
    private UpdateHandler updater;

    @Autowired
    private UpdateByQueryHandler updateByQueryHandler;

    @Autowired
    private NotifyService notifyService;


    @Autowired
    private TaskDao taskDao;

    @Value("${notify.delta.threshold.cnt:5000}")
    private String thresholdOnMsgCnt;

    @Value("${notify.dalta.threshold.time:300000}")
    private String thresholdOnTime;

    @Value("${notify.weixin.users:dengchao}")
    private String defaultWatcher;

    private static final Class<HJIndexRequest> indexClz = HJIndexRequest.class;

    private static final Class<HJPartialUpdateRequest> updateClz = HJPartialUpdateRequest.class;

    private static final Class<HJDeleteRequest> deleteClz = HJDeleteRequest.class;

    private static final Class<HJUpdateByQueryRequest> updateByQueryClz = HJUpdateByQueryRequest.class;

    private static final String INDEX_IDENTIER = "source";

    private static final String REQUEST_TYPE = "requestType";

    private static final String UPDATEBYQUERY_IDENTIER = "script";

    private Cache<String, TreeSet<Date>> failureTimeCache;

    private LoadingCache<String, String> watcherCache;

    @PostConstruct
    void initialCache() {

        // 设置缓存过期的时间为发送间隔的阈值加1
        failureTimeCache = CacheBuilder.newBuilder()
                .expireAfterWrite(Integer.valueOf(thresholdOnMsgCnt) + 1, TimeUnit.MINUTES).build();

        watcherCache = CacheBuilder.newBuilder().expireAfterWrite(5, TimeUnit.MINUTES)
                .build(new CacheLoader<String, String>() {
                    @Override
                    public String load(String index) throws Exception {
                        return taskDao.getOwner(index);
                    }
                });
    }


    @Override
    public void onMessage(Message message) {

        MDC.put("requestId", IpUtil.getIp());
        MDC.put("serverIp", IpUtil.getIp());

        String contentType = message.getMessageProperties().getContentType();
        Object req = null;

        try {
            for (MessageType msgType : MessageType.values()) {
                if (msgType.getContentType().equals(contentType)) {
                    req = msgType.getConverter().fromMessage(message);
                    break;
                }
            }
            String msgId = message.getMessageProperties().getMessageId();
            log.info("Receive msgid: {}, request body : {} ", msgId, req);

            if (req instanceof JSONObject) {
                JSONObject json = (JSONObject) req;
                handleJsonRequest(json);
            } else if (req instanceof String) {
                JSONObject json = JSON.parseObject(req.toString());
                handleJsonRequest(json);
            } else if (req instanceof HJIndexRequest) {
                indexer.handle((HJIndexRequest) req);
            } else if (req instanceof HJDeleteRequest) {
                deleter.handle((HJDeleteRequest) req);
            } else if (req instanceof HJUpdateRequest) {
                updater.handle((HJUpdateRequest) req);
            } else if (req instanceof HJUpdateByQueryRequest) {
                Object resp = updateByQueryHandler.handle((HJUpdateByQueryRequest) req);
                if (StringUtils.isNotBlank(((HJUpdateByQueryRequest) req).getCallback())) {
                    // todo: callback优化
                    try {
                        HttpClientUtil.sendHttpPost(((HJUpdateByQueryRequest) req).getCallback(),
                                JSONObject.toJSONString(resp));
                    } catch (Exception ex) {
                        log.error("HJUpdateByQueryRequest encounter exception", ex);
                    }
                }
            } else {
                String errorMsg = "Cannot determine handler for the msg [" + msgId + "], req:" + req;
                log.error(errorMsg);
                throw new ConsumeMessageException(errorMsg);
            }
        } catch (Exception e) {
            try {
                notifyFailure(req, "Fail to consume message:" + req + e.getMessage());
            } catch (Exception ex) {
                log.error("enconter exception ", ex);
            }
            log.error("Fail handler msg, msg: " + req, e);
        }
    }

    private void handleJsonRequest(JSONObject json) {
        if (json.containsKey(REQUEST_TYPE) && json.getInteger(REQUEST_TYPE) == RequestType.UPDATE.getValue()) {
            HJPartialUpdateRequest updateReq = JSONObject.toJavaObject(json, updateClz);
            updater.handle(updateReq);
        } else if ((json.containsKey(REQUEST_TYPE) && json.getInteger(REQUEST_TYPE) == RequestType.INDEX.getValue())
                || json.containsKey(INDEX_IDENTIER)) {
            HJIndexRequest indexReq = JSONObject.toJavaObject(json, indexClz);
            indexer.handle(indexReq);
        } else if ((json.containsKey(REQUEST_TYPE) && json.getInteger(REQUEST_TYPE) == RequestType.UPDATEBYQUERY.getValue())
                || json.containsKey(UPDATEBYQUERY_IDENTIER)) {
            HJUpdateByQueryRequest updateByQueryReq = JSONObject.toJavaObject(json, updateByQueryClz);
            updateByQueryHandler.handle(updateByQueryReq);
        } else {
            deleter.handle(JSONObject.toJavaObject(json, deleteClz));
        }
    }

    private void notifyFailure(Object req, String msg) {

        // 获取请求的索引
        String index = getIndex(req);
        if (StringUtils.isBlank(index)) {
            return;
        }

        // 获取索引的owner
        String watcher = StringUtils.EMPTY;

        // 获取已经发送失败的记录的时间戳
        TreeSet<Date> dates = failureTimeCache.getIfPresent(index);

        // 标记当前时间节点
        Date now = new Date();

        if (CollectionUtils.isEmpty(dates)) {

            // 如果是第一次失败，直接发送消息
            notifyService.inform(watcher, msg);
        } else if (dates.size() > Integer.valueOf(thresholdOnMsgCnt)) {

            // 如果失败的消息总量超过阈值，则发送消息，并清空缓存，防止OOM
            notifyService.inform(watcher, "已有" + thresholdOnMsgCnt + "消息添加到索引" + index + "失败，请及时处理.");
            cleanCache(index, now);
        } else {

            // 如果距上一次发送警告的时间超过阈值，则再次发送。并清空缓存
            long timeDiff = now.getTime() - dates.first().getTime();
            if (timeDiff > Integer.valueOf(thresholdOnTime)) {
                notifyService.inform(watcher,
                        "在过去的" + thresholdOnTime + "分钟内，共有" + dates.size() + "消息添加到索引" + index + "失败，请及时处理。");
                cleanCache(index, now);
            }
        }

        // 把本条失败记录放入缓存
        cacheFailure(index, now);
    }

    private void cacheFailure(String index, Date now) {
        TreeSet<Date> dates;
        dates = failureTimeCache.getIfPresent(index);
        dates = dates == null ? new TreeSet<>() : dates;
        dates.add(now);
        failureTimeCache.put(index, dates);
    }

    private String getWatcher(String index) {
        String watcher = defaultWatcher;
        try {
            watcher = watcher + "|" + watcherCache.get(index);
        } catch (Exception e) {
            log.error("Failed to get owner of index: " + index + ", notify master instead.", e);
        }
        return watcher;
    }

    private String getIndex(Object req) {
        if (req instanceof JSONObject) {
            return ((JSONObject) req).getString("index");
        }

        if (req instanceof HJEsRequest) {
            return ((HJEsRequest) req).getIndex();
        }
        log.error("找不到索引，发送消息失败.");

        return null;
    }

    private void cleanCache(String index, Date now) {
        // 清空数据， 防止短时间类大量数据进入缓存，引起OOM
        TreeSet<Date> set = new TreeSet<>();
        failureTimeCache.put(index, set);
    }

}
